﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Security.Cryptography;
using System.Text;
using System.Threading.Tasks;
using System.Timers;
using uPLibrary.Networking.M2Mqtt;
using uPLibrary.Networking.M2Mqtt.Messages;

namespace SDWXServer.Device
{
    public class SDMqttClient
    {
        protected System.Timers.Timer TimerDataService { get; private set; }
        public MqttClient MqttClient { get; private set; }
        public string HostName { get; private set; }
        public Int32 Port { get; private set; }
        public string UserName { get; private set; }
        public string PassWord { get; private set; }

        public string ClientId { get; private set; }
        public string DeviceId { get; private set; }
        public string ParentTopic { get; set; }
        public string GroupId { get; set; }
        public List<string> SubscribeTopics { get; private set; }
        public DateTime NextCheckKeepAliveTime { get; set; }
        public ushort NextCheckKeepAliveMessageId { get; set; }
        public ConcurrentDictionary<string, DateTime> ResponsedCommands { get; set; }

        void SetClientId()
        {
            this.ClientId = string.Format("{0}@@@{1}", GroupId, DeviceId);

            if (this.SubscribeTopics.Contains(this.ClientId) == false)
            {
                this.SubscribeTopics.Add(this.ClientId);
            }
        }

        public SDMqttClient(string deviceId, string hostName, Int32 port, string userName, string passWord)
        {
            this.DeviceId = deviceId;

            this.HostName = hostName;
            this.Port = port;
            this.UserName = userName;
            this.PassWord = passWord;

            this.SubscribeTopics = new List<string>();
            this.NextCheckKeepAliveTime = DateTime.Now.AddMinutes(10d);
            this.ResponsedCommands = new ConcurrentDictionary<string, DateTime>();

            NewMqttClient();

            #region 定时器检查服务
            this.TimerDataService = new System.Timers.Timer(60000);
            this.TimerDataService.Elapsed += new ElapsedEventHandler(OnTimedHandlerEvent);
            this.TimerDataService.Enabled = false;
            #endregion
        }
        protected void NewMqttClient()
        {
            this.MqttClient = new MqttClient(HostName, Port, false, null, null, MqttSslProtocols.None);
            this.MqttClient.MqttMsgPublishReceived += OnMqttMsgPublishReceived;
            this.MqttClient.MqttMsgPublished += OnMqttMsgPublished;
            this.MqttClient.ConnectionClosed += OnConnectionClosed;
        }
        public void Start()
        {
            try
            {
                SetClientId();
                Connect();
            }
            catch (Exception e)
            {
                KD.Logger.Error("消息队列服务连接失败", e);
            }
            finally
            {
                this.TimerDataService.Enabled = true;
            }
        }
        public void Stop()
        {
            try
            {
                this.TimerDataService.Enabled = false;

                this.Disconnect(false);
            }
            catch (Exception e)
            {

            }
        }
        public void Disconnect(bool isNewMqttClient)
        {
            try
            {
                this.MqttClient.MqttMsgPublishReceived -= OnMqttMsgPublishReceived;
                this.MqttClient.MqttMsgPublished -= OnMqttMsgPublished;
                this.MqttClient.ConnectionClosed -= OnConnectionClosed;

                if (this.MqttClient.IsConnected)
                {
                    this.MqttClient.Disconnect();
                }
            }
            catch (Exception e)
            {
                KD.Logger.Info(string.Format("消息队列服务断开失败:{0}", e.Message));
            }

            try
            {
                if (isNewMqttClient)
                {
                    NewMqttClient();
                }
            }
            catch (Exception e)
            {
                KD.Logger.Info(string.Format("消息队列服务NewMqttClient:{0}", e.Message));
            }
        }
        private void OnTimedHandlerEvent(object sender, ElapsedEventArgs e)
        {
            try
            {
                if (this.MqttClient.IsConnected)
                {
                    if (DateTime.Now > this.NextCheckKeepAliveTime)
                    {
                        this.NextCheckKeepAliveTime = DateTime.Now.AddMinutes(10d);

                        if (this.NextCheckKeepAliveMessageId > 0)
                        {
                            this.NextCheckKeepAliveMessageId = 0;

                            this.Disconnect(true);
                            KD.Logger.Info("消息队列服务KeepAlive - Disconnect - OnTimedHandlerEvent");

                            Connect();
                        }
                        else
                        {
                            var subTopic = this.ClientId;
                            var topic = string.Format("{0}/{1}", ParentTopic, subTopic);
                            var message = "{}";
                            this.NextCheckKeepAliveMessageId = this.MqttClient.Publish(topic, Encoding.UTF8.GetBytes(message), MqttMsgBase.QOS_LEVEL_AT_LEAST_ONCE, false);
                        }
                    }

                    return;
                }

                Connect();
            }
            catch (Exception exp)
            {
                KD.Logger.Error("OnTimedHandlerEvent", exp);
            }
        }
        private void Connect()
        {
            try
            {
                var passWord = this.PassWord;
                if (this.HostName.EndsWith("mqtt.aliyuncs.com", StringComparison.OrdinalIgnoreCase))
                {
                    passWord = HMACSHA1(this.PassWord, this.ClientId);
                }

                this.MqttClient.Connect(this.ClientId, this.UserName, passWord, true, 30);

                this.Subscribe(this.SubscribeTopics);

                KD.Logger.Info("消息队列服务连接");
            }
            catch (Exception exp)
            {
                KD.Logger.Error("消息队列服务连接失败", exp);
            }
        }

        private string HMACSHA1(string key, string dataToSign)
        {
            Byte[] secretBytes = UTF8Encoding.UTF8.GetBytes(key);
            HMACSHA1 hmac = new HMACSHA1(secretBytes);
            Byte[] dataBytes = UTF8Encoding.UTF8.GetBytes(dataToSign);
            Byte[] calcHash = hmac.ComputeHash(dataBytes);
            String calcHashString = Convert.ToBase64String(calcHash);
            return calcHashString;
        }

        public void Subscribe(List<string> subTopics)
        {
            if (subTopics == null || subTopics.Count == 0) return;
            foreach (var topic in subTopics)
            {
                if (this.SubscribeTopics.Contains(topic)) continue;

                this.SubscribeTopics.Add(topic);
            }

            var subTopicArray = subTopics.Select(p => string.Format("{0}/{1}", ParentTopic, p)).ToArray();
            var qosLevels = subTopics.Select(p => MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE).ToArray();

            this.MqttClient.Subscribe(subTopicArray, qosLevels);
        }
        public void Publish(string subTopic, string message)
        {
            try
            {
                if (this.MqttClient.IsConnected == false) return;
                if (string.IsNullOrWhiteSpace(subTopic)) return;

                var topic = string.Format("{0}/{1}", ParentTopic, subTopic);
                this.MqttClient.Publish(topic, Encoding.UTF8.GetBytes(message), MqttMsgBase.QOS_LEVEL_AT_LEAST_ONCE, false);
            }
            catch (Exception exp)
            {
                KD.Logger.Error("Publish", exp);
            }
        }
        private void OnConnectionClosed(object sender, EventArgs e)
        {
            KD.Logger.Info("消息队列服务连接断开");
        }

        private void OnMqttMsgPublished(object sender, MqttMsgPublishedEventArgs e)
        {
            if (this.NextCheckKeepAliveMessageId > 0)
            {
                if (this.NextCheckKeepAliveMessageId == e.MessageId)
                {
                    if (e.IsPublished == false)
                    {
                        KD.Logger.Info("消息队列服务KeepAlive - IsPublished - false");
                    }
                    else
                    {
                        this.NextCheckKeepAliveMessageId = 0;

                        KD.Logger.Info("消息队列服务KeepAlive - OK");
                    }
                }
            }
        }

        private void OnMqttMsgPublishReceived(object sender, MqttMsgPublishEventArgs e)
        {
            if (this.OnMqttMsgPublishReceivedAction != null)
            {
                var message = Encoding.UTF8.GetString(e.Message);
                if (this.NextCheckKeepAliveMessageId > 0)
                {
                    if (message == "{}" && e.Topic != null && e.Topic.Contains(this.ClientId))
                    {
                        this.NextCheckKeepAliveMessageId = 0;

                        KD.Logger.Info("消息队列服务KeepAlive - OK - {}");
                    }
                }

                var arg = new SDMqttMsgPublishEventArgs(this, this.DeviceId, e.Topic, message, e.DupFlag, e.QosLevel, e.Retain);

                this.OnMqttMsgPublishReceivedAction(arg);
            }
        }
        public Action<SDMqttMsgPublishEventArgs> OnMqttMsgPublishReceivedAction { get; set; }
    }
}